/*
 * Copyright (c) 2010 CCX(China) Co.,Ltd. All Rights Reserved.
 *
 * This software is the confidential and proprietary information of
 * CCX(China) Co.,Ltd. ("Confidential Information").
 * It may not be copied or reproduced in any manner without the express 
 * written permission of CCX(China) Co.,Ltd.
 * 
 *	Created on 2010-12-22 下午03:07:59
 */
package cn.com.ccxe.core.crawl;

import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

import org.apache.commons.io.IOUtils;
import org.apache.http.Header;
import org.apache.http.HeaderElement;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.HttpVersion;
import org.apache.http.NameValuePair;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.params.AllClientPNames;
import org.apache.http.conn.params.ConnManagerParams;
import org.apache.http.conn.params.ConnPerRouteBean;
import org.apache.http.conn.params.ConnRoutePNames;
import org.apache.http.conn.scheme.PlainSocketFactory;
import org.apache.http.conn.scheme.Scheme;
import org.apache.http.conn.scheme.SchemeRegistry;
import org.apache.http.conn.ssl.SSLSocketFactory;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.HttpParams;
import org.apache.http.params.HttpProtocolParamBean;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import cn.com.ccxe.core.entity.Link;
import cn.com.ccxe.core.entity.WebData;
import cn.com.ccxe.core.util.CodeUtil;
import cn.com.ccxe.core.util.Configurations;

/**
 * 数据采集器。 封装了{@link HttpClient}，发出HTTP请求和处理HTTP响应，并提供相应的请求控制。
 * 
 * 
 * @author hetao
 */
public final class Fetcher implements AllClientPNames {

	private static DefaultHttpClient httpClient;
	private static ThreadSafeClientConnManager connectionManager;
	private static int processedCount = 0;
	private static long startOfPeriod = 0;
	private static long lastFetchTime = 0;
	public static final String CHARSET = "GBK";
	private static long politenessDelay = Configurations.getInt("fetcher.default_politeness_delay", 200);
	private static final Logger logger = Logger.getLogger(Fetcher.class);
	public static final int MAX_DOWNLOAD_SIZE = Configurations.getInt("fetcher.max_download_size", 1048576);

	public static long getPolitenessDelay() {
		return politenessDelay;
	}

	public static void setPolitenessDelay(long politenessDelay) {
		Fetcher.politenessDelay = politenessDelay;
	}

	/**
	 * 抓取，抓取时会改变传入的WebURL的一些状态，如抓取时间。
	 * 
	 * @param webUrl
	 * @return
	 */
	public WebData fetch(Link link) throws IOException {
		logger.setLevel(Level.ERROR);
		String toFetchURL = link.getPath();
		HttpGet getMethod = null;
		WebData data = new WebData(link);
		try {
			checkFetchOfPeriod();

			getMethod = initGetMethod(toFetchURL);
			HttpResponse response = httpClient.execute(getMethod);
			HttpEntity entity = response.getEntity();

			int statusCode = response.getStatusLine().getStatusCode();
			link.setStatusCode(statusCode);
			logger.info("Response content length: " + entity.getContentLength());
			logger.info("Chunked?: " + entity.isChunked());
			switch (statusCode / 100) {
			case HttpStatus.SC_CONTINUE:
				logger.info("协议转换: " + toFetchURL);
				break;
			case HttpStatus.SC_OK / 100:

				String charset = getCharset(entity);
				if (charset == null || "".equals(charset)) {
					charset = "GB18030";
				}
				data.setHtml(getContent(entity, charset));

				StringBuilder sb = new StringBuilder();
				sb.append("【采集成功: ").append(statusCode).append("\t").append("Content-Length:").append(
						response.getEntity().getContentLength()).append("\tHTML-Length:").append(
						data.getHtml().length()).append("】\n").append("URL:").append(toFetchURL);
				logger.info(sb.toString());
				link.setFetchTime(CodeUtil.getDate());
				break;
			case HttpStatus.SC_MULTIPLE_CHOICES / 100:
				logger.info("页面被重定向：" + toFetchURL);
				break;
			case HttpStatus.SC_NOT_FOUND / 100:
				logger.info("页面未发现：" + toFetchURL);
				break;
			case HttpStatus.SC_BAD_GATEWAY / 100:
				logger.info("服务器内部错误：" + toFetchURL);
				break;
			default:
				logger.info("未知错误");
				break;
			}
		} catch(SocketTimeoutException e) {
			logger.info("抓取延时...");
			throw e;
		} catch (IOException e) {
			StringBuilder sb = new StringBuilder();
			sb.append("Fatal transport error: ").append(e.getMessage())
			  .append(" while fetching ").append(toFetchURL).append(")");
			logger.error(sb.toString());
			throw e;
		} catch (Exception e) {
			if (e.getMessage() == null) {
				logger.info("Error while fetching " + toFetchURL);
				e.printStackTrace();
			} else {
				logger.info(e.getMessage() + " while fetching " + toFetchURL);
				e.printStackTrace();
			}
		} finally {
			clear(getMethod, null);
		}
		return data;
	}

	private String getContent(HttpEntity entity, String charset) throws IllegalStateException, IOException {

		if (entity != null) {
			InputStream instream = entity.getContent();
			try {
				return IOUtils.toString(instream,CHARSET);
			} catch (IOException ex) {
				throw ex;
			} finally {
				instream.close();
			}
		}
		return "";
	}

	private String getCharset(HttpEntity entity) {
		Header header = entity.getContentType();
		for (HeaderElement he : header.getElements()) {
			if (he != null) {
				// 获取Content-Type中的Charset key-vaule pair
				NameValuePair pair = he.getParameterByName("charset");
				if (pair != null && "charset".equals(pair.getName())) {
					return pair.getValue();
				}
			}
		}
		return "";
	}

	/**
	 * 抓取频率控制。
	 * 
	 */
	private synchronized void checkFetchOfPeriod() {
		long now = (new Date()).getTime();
		if (now - startOfPeriod > 10000) {
			long fetchedPerSecond = processedCount / ((now - startOfPeriod) / 1000);
			logger.info("Number of pages fetched per second: " + fetchedPerSecond);
			processedCount = 0;
			startOfPeriod = now;
		}
		processedCount++;
		if (now - lastFetchTime < politenessDelay) {
			sleep(politenessDelay - (now - lastFetchTime));
		}
		lastFetchTime = (new Date()).getTime();
	}

	private void sleep(long time) {
		try {
			TimeUnit.MILLISECONDS.sleep(time);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private static void clear(HttpGet get, HttpEntity entity) {
		try {
			if (entity != null) {
				entity.consumeContent();
			} else if (get != null) {
				get.abort();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 初始化get方法。
	 * 
	 * @param toFetchURL
	 * @return
	 */
	private static HttpGet initGetMethod(String toFetchURL) {
		HttpGet getMethod = new HttpGet(toFetchURL);
		getMethod.addHeader("Accept-Language", "zh-CN,zh");
		// getMethod.addHeader("Accept-Encoding",
		// "deflate, gzip, x-gzip, identity, *;q=0");
		// getMethod.addHeader("Accept-Charset","zh-CN,zh;q=0.9,en;q=0.8");
		return getMethod;
	}

	static {
		HttpParams params = new BasicHttpParams();
		HttpProtocolParamBean paramsBean = new HttpProtocolParamBean(params);
		paramsBean.setVersion(HttpVersion.HTTP_1_1);
		paramsBean.setContentCharset("GBK");
		paramsBean.setUseExpectContinue(false);

		params.setParameter(USER_AGENT, Configurations.getString("fetcher.user_agent", "Opera 10.04"));
		params.setIntParameter(SO_TIMEOUT, Configurations.getInt("fetcher.socket_timeout", 30000));
		params.setIntParameter(CONNECTION_TIMEOUT, Configurations.getInt("fetcher.connection_timeout", 35000));
		params.setBooleanParameter(HANDLE_REDIRECTS, Configurations.getBoolean("fetcher.follow_redirects", true));

		ConnPerRouteBean connPerRouteBean = new ConnPerRouteBean();
		connPerRouteBean.setDefaultMaxPerRoute(Configurations.getInt("fetcher.max_connections_per_host", 100));

		ConnManagerParams.setMaxConnectionsPerRoute(params, connPerRouteBean);
		ConnManagerParams.setMaxTotalConnections(params, Configurations.getInt("fetcher.max_total_connections", 100));

		SchemeRegistry schemeRegistry = new SchemeRegistry();
		schemeRegistry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));

		if (Configurations.getBoolean("fetcher.crawl_https", false)) {
			schemeRegistry.register(new Scheme("https", SSLSocketFactory.getSocketFactory(), 443));
			logger.info("使用HTTPS协议");
		}
		connectionManager = new ThreadSafeClientConnManager(params, schemeRegistry);
		httpClient = new DefaultHttpClient(connectionManager, params);
	}

	public static void setProxy(String proxyHost, int proxyPort) {
		HttpHost proxy = new HttpHost(proxyHost, proxyPort);
		httpClient.getParams().setParameter(ConnRoutePNames.DEFAULT_PROXY, proxy);
	}

	public static void setProxy(String proxyHost, int proxyPort, String username, String password) {
		httpClient.getCredentialsProvider().setCredentials(new AuthScope(proxyHost, proxyPort),
				new UsernamePasswordCredentials(username, password));
		setProxy(proxyHost, proxyPort);
	}

}
